package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo21Cache {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("cache")
    val sc = new SparkContext(conf)

    //指定checkpoint地址
    sc.setCheckpointDir("data/checkpoint")


    //读取数据
    val studentsRDD: RDD[String] = sc.textFile("data/students.txt")


    /**
     * RDD中默认没有数据，缓存级别默认为NONE( StorageLevel.NONE)
     */
    println(studentsRDD.getStorageLevel)

    //提取字段
    val stuRDD: RDD[(String, String, Int, String, String)] = studentsRDD.map(line => {
      println("stuRDD")
      val split: Array[String] = line.split(",")
      val id: String = split(0)
      val name: String = split(1)
      val age: Int = split(2).toInt
      val sex: String = split(3)
      val clazz: String = split(4)
      (id, name, age, sex, clazz)
    })

    /**
     * 默认情况下，入股一个RDD被多次使用，会计算多次
     *
     * 缓存级别
     * StorageLevel.MEMORY_ONLY： 仅内存
     * StorageLevel.MEMORY_AND_DISK_SER: 将数据压缩之后方内，内存放不下在方磁盘，
     * 为甚不选择MEMORY_AND_DISK：应该不管压缩不压缩，内存都比磁盘快，所以尽量将数据压缩之后放内存
     *
     * 缓存级别选择和数据量有关，如果内存放不下选择使用MEMORY_AND_DISK_SER
     */


    //stuRDD.persist(StorageLevel.MEMORY_ONLY)
    stuRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)


    //MEMORY_ONLY:的简写
    //stuRDD.cache()


    /**
     * 将RDD的数据缓存到hdfs中，如果任务失败重启。缓存的数据不会丢失
     * checkpoint在离线数据处理中一般不用。
     * 在后面的Spark Streaming中需要使用
     */
    stuRDD.checkpoint()


    /**
     * 统计班级的人数
     */

    stuRDD
      .map(stu => (stu._5, 1))
      .reduceByKey((x, y) => x + y)
      .foreach(println)

    /**
     * 统计性别人数
     */

    stuRDD
      .map(stu => (stu._4, 1))
      .reduceByKey((x, y) => x + y)
      .foreach(println)

    /**
     * 统计年龄的人数
     */

    stuRDD
      .map(stu => (stu._3, 1))
      .reduceByKey((x, y) => x + y)
      .foreach(println)


  }

}
